package com.lianjia.sh.se.kafka.client;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.text.MessageFormat;
import java.util.Arrays;
import java.util.Properties;

/**
 * kafka监听器超类，用于接收服务器端的消息
 * Created by yangsp1 on 16/7/27.
 */
public abstract class AbstractKafkaListener {
    Logger logger = LoggerFactory.getLogger(AbstractKafkaListener.class);
    @Autowired
    KafkaConsumerFactory kafkaConsumerFactory;

    //主题名
    private String topic;

    //间隔多久拉取一次（单位：毫秒），是自己加的。
    private Long lingerMs = 2000L;

    //消费者自定义配置
    private Properties configs = new Properties();

    /**
     *  消费消息
     * @param value
     */
    public abstract void onMessage(String value);

    /**
     * 定时拉取数据
     */
    public void startFrequentPollRecords(){
        new Thread(){
            @Override
            public void run() {
                Consumer consumer = kafkaConsumerFactory.createConsumer(this, configs);
                consumer.subscribe(Arrays.asList(topic));
                while (true){
                    System.out.println("aaaaaaa");
                    try {
                        ConsumerRecords<String, String> records = consumer.poll(100);
                        for (ConsumerRecord<String, String> record : records) {
                            onMessage(record.value());
                            logger.info(MessageFormat.format("offset = {0}, key = {1}, value = {2}", record.offset(), record.key(), record.value()));
                        }
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                    try {
                        Thread.sleep(lingerMs);
                    } catch (Exception e) {
                        logger.error(e.getMessage(), e);
                    }
                }
            }
        }.start();
    }



    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public Properties getConfigs() {
        return configs;
    }

    public Long getLingerMs() {
        return lingerMs;
    }

    public void setLingerMs(Long lingerMs) {
        this.lingerMs = lingerMs;
    }

    public KafkaConsumerFactory getKafkaConsumerFactory() {
        return kafkaConsumerFactory;
    }

    public void setKafkaConsumerFactory(KafkaConsumerFactory kafkaConsumerFactory) {
        this.kafkaConsumerFactory = kafkaConsumerFactory;
    }
}
